package com.chandon.juc.threadPool;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.stream.IntStream;

/**
 * WorkStealingPool与别的线程池不同，基于ForkjoinPool构建。
 * ThreadPoolExecutor构建的线程池，所有线程共用一个阻塞队列存放任务。
 * ForkjoinPool特点：
 * 中每一个线程都有自己的阻塞队列，可以将一个大的任务拆分为几个小任务并存在阻塞队列中。
 * 线程池中的空闲线程可以去处理有任务的阻塞队列中的任务，提升效率。
 * 并不是所有任务都可以拆分，首先任务要足够大，耗时足够长。
 * 使用时不推荐使用Runable，Callbale，使用RecursiveAction，RecursiveTask代替。
 *
 * @author Chandon
 */

public class TestWorkStealingPool {

	public static void main(String[] args) throws ExecutionException, InterruptedException {

		ForkJoinPool workStealingPool = (ForkJoinPool) Executors.newWorkStealingPool(10);

		int[] array = IntStream.rangeClosed(0, 100000).toArray();

		System.out.println(workStealingPool.submit(new SumRecursiveTask(0, array.length - 1, array, 10)).get());
	}


}

class SumRecursiveTask extends RecursiveTask<Integer> {
	// 单个任务处理的阈值
	private final static int THRESHOLD = 20;
	private int start;
	private int end;
	private int stride;
	private int[] arr;

	public SumRecursiveTask(int start, int end, int[] arr, int stride) {
		this.start = start;
		this.end = end;
		this.arr = arr;
		this.stride = stride;
	}


	@Override
	public Integer compute()  {
		if (end - start >= stride) {
			System.out.println("任务拆分...");
			// 对任务进行切分
			SumRecursiveTask task1 = new SumRecursiveTask(start, (end + start) / 2, arr, stride);
			SumRecursiveTask task2 = new SumRecursiveTask((end + start) / 2 + 1, end, arr, stride);

			invokeAll(task1,task2);
			return task1.join()+task2.join();
		}
		// 直接处理 返回结果
		int sum = 0;
		for (int i = start; i <= end; i++) {
			sum += arr[i];
		}
		System.out.println(Thread.currentThread().getName() + "线程执行情况:" + start + "~" + end + ":" + sum);
		return sum;
	}
}
